package com.test.common.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;

import java.util.*;

@Slf4j
public class KafkaUtils {

    /**
     * 创建一些topic
     *
     * @param brokerUrl kafka地址：localhost:9092
     * @param topics    需要创建的topic，分区1副本1
     * @return 创建结果
     */
    public static KafkaFuture<Void> createTopics(String brokerUrl, String[] topics) {

        log.info("创建topic|{}", topics[0]);

        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
        AdminClient adminClient = AdminClient.create(properties);

        List<NewTopic> newTopics = new ArrayList<>();

        for (String topic : topics) {

            NewTopic newTopic = new NewTopic(topic, 1, (short) 1);

            newTopics.add(newTopic);
        }

        CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);

        adminClient.close();

        try {
            Thread.sleep(1000);
        } catch (Exception e) {
            log.error("REPLACE ME|{}", e.getMessage());
        }

        return createTopicsResult.all();
    }

    /**
     * 查询所有的topic名称
     */
    public static Set<String> getAllTopics(String brokerUrl) {

        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
        AdminClient adminClient = AdminClient.create(properties);

        ListTopicsResult topicRequestor = adminClient.listTopics();

        KafkaFuture<Set<String>> kafkaFuture = topicRequestor.names();

        adminClient.close();

        Set<String> resSet = null;

        try {
            resSet = kafkaFuture.get();
        } catch (Exception e) {
            log.error("REPLACE ME|{}", e.getMessage());
        }

        return resSet;
    }

    public static TopicDescription getTopicInfo(String brokerUrl, String topic) {

        log.info("查询topic|{}", topic);

        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
        AdminClient adminClient = AdminClient.create(properties);

        List<String> topicList = new ArrayList<>();

        topicList.add(topic);

        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topicList);

        KafkaFuture<Map<String, TopicDescription>> kafkaFuture = describeTopicsResult.all();

        adminClient.close();

        TopicDescription resTopicDesc = null;

        try {
            resTopicDesc = kafkaFuture.get().get(topic);
        } catch (Exception e) {
            log.error("REPLACE ME|{}", e.getMessage());
        }

        return resTopicDesc;
    }

    /**
     * 删除topic
     * 需要在kafka的配置文件中将auto.create.topics.enable设置为false、delete.topic.enable设置为true（没有相关配置项的话就直接添加）
     */
    public static KafkaFuture<Void> deleteTopics(String brokerUrl, String[] topics) {

        log.info("删除topic|{}", topics[0]);

        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
        AdminClient adminClient = AdminClient.create(properties);

        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(topics));

        adminClient.close();

        try {
            Thread.sleep(1000);
        } catch (Exception e) {
            log.error("REPLACE ME|{}", e.getMessage());
        }

        return deleteTopicsResult.all();
    }

    public static void main(String[] args) {

		System.out.println("清理TOPIC：" + deleteTopics("172.18.84.143:9092", new String[]{"test"}));
        System.out.println("创建TOPIC：" + createTopics("172.18.84.143:9092", new String[]{"test"}));
        System.out.println("查询TOPIC：" + getTopicInfo("172.18.84.143:9092", "test"));
    }
}